// -*-mode:c++; coding:utf-8-*-


#include <boost/foreach.hpp>

#include <xthread/guard.hpp>

#include "kv_scanner_manager.hpp"

namespace xbase {
	kv_scanner_manager::kv_scanner_manager ()
	{
	}

	kv_scanner_manager::~kv_scanner_manager ()
	{
		boost::mutex::scoped_lock lock(m_mutex);
		m_scanner_map.clear();
		m_tablet_scanner_map.clear();
	}

	void kv_scanner_manager::add_scanner (const tablet_scanner::pointer &p_scanner) {
		boost::mutex::scoped_lock lock(m_mutex);
		m_scanner_map[p_scanner->id()] = p_scanner;
		m_tablet_scanner_map.insert(tablet_scanner_map::value_type(p_scanner->tablet_name(),
									   p_scanner->id()));
	}

	tablet_scanner::pointer kv_scanner_manager::get_scanner (const uint64_t &p_id) const {
		boost::mutex::scoped_lock lock(m_mutex);
		tablet_scanner::pointer s;
		const scanner_map_type::const_iterator iter = m_scanner_map.find(p_id);
		if(iter != m_scanner_map.end())
		{
			s = iter->second;
		}
		return s;
	}

	void kv_scanner_manager::close_scanner (const uint64_t &p_id) {
		boost::mutex::scoped_lock lock(m_mutex);
		const scanner_map_type::iterator iter = m_scanner_map.find(p_id);
		if(iter != m_scanner_map.end())
		{
			std::pair<tablet_scanner_map::iterator, tablet_scanner_map::iterator>
				range = m_tablet_scanner_map.equal_range(iter->second->tablet_name());
			for(;range.first != range.second; ++range.first)
			{
				if(range.first->second == p_id)
				{
					break;
				}
			}
			if(range.first != range.second) // found, and erase
			{
				m_tablet_scanner_map.erase(range.first);
			}
			// then erase iter
			m_scanner_map.erase(iter);
			
			LOG_IF((range.first == range.second),
			       WARN,
			       "scanner " << p_id << " not found in m_tablet_scanner_map");
		}
	}

	// TODO: 假如Scanner正在使用，这样关闭Scanner应该是没有问题，并且不影响使用。
	//
	// 最好在这里对Scanner加锁，关闭后设置为超时；在next操作中，获
	// 取锁后，再检查是否已经关闭了。但这会导致因scanner超时导致返
	// 回eos，从而使得客户端以为扫描结束了。
	//
	// 暂时采用当前的实现吧，如果出现问题了，再考虑完善。
	void kv_scanner_manager::close_expired_scanners (const uint32_t &p_interval) {
		// find scanner to close
		boost::mutex::scoped_lock lock(m_mutex);
		std::vector<uint64_t> _expired_id;
		BOOST_FOREACH(const scanner_map_type::value_type &i, m_scanner_map) {
			if (i.second->is_expired(p_interval)) {
				_expired_id.push_back(i.first);
			}
		}
		lock.unlock();
		
		BOOST_FOREACH(uint64_t _id, _expired_id) {
			this->close_scanner(_id);
		}

		LOG_IF((! _expired_id.empty()),
		       DEBUG,
		       "kv_scanner_manager::close_expired_scanners "
		       << _expired_id.size() << " scanners expired and closed");
	}

	void kv_scanner_manager::close_scanner_on_tablet(const std::string &tablet_name)
	{
		// find scanner to close
		boost::mutex::scoped_lock lock(m_mutex);
		const std::pair<tablet_scanner_map::iterator, tablet_scanner_map::iterator>
			range = m_tablet_scanner_map.equal_range(tablet_name);
		LOG_ENV(std::size_t log_env_counter = 0);
		for(tablet_scanner_map::iterator it = range.first;
		    it != range.second; ++it)
		{
			const scanner_map_type::iterator end = m_scanner_map.end();
			const scanner_map_type::iterator iter = m_scanner_map.find(it->second);
			if(iter != end)
			{
				m_scanner_map.erase(iter);
			}
			LOG_ENV(++log_env_counter);
			LOG_IF((iter == end),
			       WARN,
			       "scanner " << it->second << " in tablet " << tablet_name
			       << " not found in m_scanner_map");
		}
		m_tablet_scanner_map.erase(range.first, range.second);
		LOG_IF((log_env_counter != 0),
		       DEBUG,
		       log_env_counter << " scanner in tablet " << tablet_name << " closed");
	}

} // namespace

